package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo08JdLogAnalysis {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo08JdLogAnalysis")
      .master("local")
      .config("spark.sql.shuffle.partitions", 1) // 默认200
      .getOrCreate()

    import org.apache.spark.sql.functions._
    import spark.implicits._

    // 读取jd_logs.txt数据
    val jdLogsDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", " ")
      .schema("user_id String,shop String")
      .load("Spark/data/jd_logs.txt")

    //    jdLogsDF.show()

    /**
     * 请统计：（使用DSL完成）
     * 1）每个店铺的UV（访客数），输出店铺名称、访客数
     * 2）每个店铺访问次数top3的访客信息，输出店铺名称、访客id、访问次数
     */

    // 1）每个店铺的UV（访客数），输出店铺名称、访客数
    jdLogsDF
      .groupBy($"shop")
      .agg(countDistinct($"user_id") as "cnt")
    //      .show()

    // 2）每个店铺访问次数top3的访客信息，输出店铺名称、访客id、访问次数
    jdLogsDF
      .groupBy($"shop", $"user_id")
      .agg(count("*") as "cnt")
      .select($"shop", $"user_id", $"cnt",
        row_number() over Window.partitionBy($"shop").orderBy($"cnt".desc) as "rn")
      .where($"rn" <= 3)
      .select($"shop", $"user_id", $"cnt", $"rn")
    //      .show()

    // 读取jd_user_visit_cnt.txt数据
    val jdUserVisitCntDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", " ")
      .schema("user_id String,date String,cnt Int")
      .load("Spark/data/jd_user_visit_cnt.txt")

    jdUserVisitCntDF.show()

    // DSL
    // 统计出每个用户每个月的访问次数以及累计访问次数
    jdUserVisitCntDF
      .select($"user_id", $"date".substr(6, 2) as "month", $"cnt")
      .groupBy($"user_id", $"month")
      .agg(sum($"cnt") as "total_cnt")
      .select($"user_id", $"month", $"total_cnt",
        sum($"total_cnt") over Window.partitionBy($"user_id").orderBy($"month") as "sum_cnt")
      .write
      .format("jdbc")
      .option("truncate", "true") // 防止表结构发生改变 默认采取drop 加 create table 进行overwrite
      .option("url", "jdbc:mysql://rm-bp1h7v927zia3t8iwho.mysql.rds.aliyuncs.com:3306/stu016?useSSL=false")
      .option("dbtable", "jd_user_cnt")
      .option("user", "shujia016")
      .option("password", "123456")
      .mode(SaveMode.Overwrite)
      .save()


  }

}
